package com.bossien.flink.redis;

import lombok.Data;
import lombok.experimental.Accessors;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.types.Row;

@Data
@Accessors
public  class FlinkRedisMapper implements RedisMapper<Tuple2<Boolean, Row>> {

    private String key;

    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.HSET, key);
    }

    @Override
    public String getKeyFromData(Tuple2<Boolean, Row> data) {

        return data.f1.getField(0).toString();
       // return data.f0.toString();
    }

    @Override
    public String getValueFromData(Tuple2<Boolean, Row> data) {
        return data.f0.toString()+","+data.f1.toString();
    }
}

